package ods

import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import java.util

import com.alibaba.fastjson.JSON
import dim.AreaInfo
import org.apache.hadoop.conf.Configuration
import org.apache.flink.configuration
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Delete, Put, Table}

/**
 * @author: yehw
 * @date: 2021/3/11 下午10:43
 * @description: 连接hbase
 */
class SinkHBase extends RichSinkFunction[util.ArrayList[TableObject]] {
  var connection: Connection = _
  var hbTable: Table = _

  /**
   * 实例化hbase
   **/
  override def open(parameters: configuration.Configuration): Unit = {
    val conf: Configuration = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "linux121,linux123")
    conf.set("hbase.zookeeper.property.clientPort", "2181")
    connection = ConnectionFactory.createConnection(conf)
    hbTable = connection.getTable(TableName.valueOf("lagou_shops"))
  }

  override def close(): Unit = {
    if (connection != null) {
      connection.close()
    }
  }

  /**
   * 处理每一条数据
   **/
  override def invoke(value: util.ArrayList[TableObject], context: SinkFunction.Context[_]): Unit = {
    value.forEach(x => {
      val dataBase = x.dataBase
      val tableName = x.tableName
      val typeInfo = x.typeInfo
      if (dataBase.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_trade_orders")) {
        if (typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: DataInfo = JSON.parseObject(x.dataInfo, classOf[DataInfo])
            insertTradeOrders(hbTable, info)
          })
        } else if (typeInfo.equalsIgnoreCase("update")) {} else if (typeInfo.equalsIgnoreCase("delete")) {}
      }
      if (dataBase.equalsIgnoreCase("dwshow") && tableName.equalsIgnoreCase("lagou_area")) {
        if (typeInfo.equalsIgnoreCase("insert")) {
          value.forEach(x => {
            val info: AreaInfo = JSON.parseObject(x.dataInfo, classOf[AreaInfo])
            insertArea(hbTable, info)
          })
        } else if (typeInfo.equalsIgnoreCase("update")) {
          value.forEach(x => {
            val info: AreaInfo = JSON.parseObject(x.dataInfo, classOf[AreaInfo])
            insertArea(hbTable, info)
          })
        } else if (typeInfo.equalsIgnoreCase("delete")) {
          value.forEach(x => {
            val info: AreaInfo = JSON.parseObject(x.dataInfo, classOf[AreaInfo])
            deleteArea(hbTable, info)
          })
        }
      }
    })
  }

  //lagou_area省份城市区域表，根据id删除数据
  def deleteArea(hbTable: Table, areaInfo: AreaInfo): Unit = {
    val delete = new Delete(areaInfo.id.getBytes)
    hbTable.delete(delete)
  }
  def insertArea(hbTable: Table, areaInfo: AreaInfo) : Unit = {
  println(areaInfo.toString)
  val put = new Put(areaInfo.id.getBytes())
    put.addColumn("f1".getBytes(), "name".getBytes(), areaInfo.name.getBytes())
    put.addColumn("f1".getBytes(), "pid".getBytes(), areaInfo.pid.getBytes())
    put.addColumn("f1".getBytes(), "sname".getBytes(), areaInfo.sname.getBytes())
    put.addColumn("f1".getBytes(), "level".getBytes(), areaInfo.level.getBytes())
    put.addColumn("f1".getBytes(), "citycode".getBytes(), areaInfo.citycode.getBytes())
    put.addColumn("f1".getBytes(), "yzcode".getBytes(), areaInfo.yzcode.getBytes())
    put.addColumn("f1".getBytes(), "mername".getBytes(), areaInfo.mername.getBytes())
    put.addColumn("f1".getBytes(), "lng".getBytes(), areaInfo.Lng.getBytes())
    put.addColumn("f1".getBytes(), "lat".getBytes(), areaInfo.Lat.getBytes())
    put.addColumn("f1".getBytes(), "pinyin".getBytes(), areaInfo.pinyin.getBytes())
    hbTable.put(put)
}

def insertTradeOrders (hbTable: Table, dataInfo: DataInfo): Unit = {
  val put = new Put (dataInfo.orderId.getBytes)
  put.addColumn ("f1".getBytes, "modifiedTime".getBytes, dataInfo.modifiedTime.getBytes () )
  put.addColumn ("f1".getBytes, "orderNo".getBytes, dataInfo.orderNo.getBytes () )
  put.addColumn ("f1".getBytes, "isPay".getBytes, dataInfo.isPay.getBytes () )
  put.addColumn ("f1".getBytes, "orderId".getBytes, dataInfo.orderId.getBytes () )
  put.addColumn ("f1".getBytes, "tradeSrc".getBytes, dataInfo.tradeSrc.getBytes () )
  put.addColumn ("f1".getBytes, "payTime".getBytes, dataInfo.payTime.getBytes () )
  put.addColumn ("f1".getBytes, "productMoney".getBytes, dataInfo.productMoney.getBytes () )
  put.addColumn ("f1".getBytes, "totalMoney".getBytes, dataInfo.totalMoney.getBytes () )
  put.addColumn ("f1".getBytes, "dataFlag".getBytes, dataInfo.dataFlag.getBytes () )
  put.addColumn ("f1".getBytes, "userId".getBytes, dataInfo.userId.getBytes () )
  put.addColumn ("f1".getBytes, "areaId".getBytes, dataInfo.areaId.getBytes () )
  put.addColumn ("f1".getBytes, "createTime".getBytes, dataInfo.createTime.getBytes () )
  put.addColumn ("f1".getBytes, "payMethod".getBytes, dataInfo.payMethod.getBytes () )
  put.addColumn ("f1".getBytes, "isRefund".getBytes, dataInfo.isRefund.getBytes () )
  put.addColumn ("f1".getBytes, "tradeType".getBytes, dataInfo.tradeType.getBytes () )
  put.addColumn ("f1".getBytes, "status".getBytes, dataInfo.status.getBytes () )
  hbTable.put (put)
}
}
